package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)
import "sync/atomic"
import "ds/labrpc"

// import "bytes"
// import "../labgob"

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

type LogEntry struct {
	Term int
	Data interface{}
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mutex     sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	myId      int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	loopMutex         sync.Mutex
	loop              bool
	ServerIDS         [][]int       //服务器列表
	applyChan         chan ApplyMsg //已经被Apply的消息
	currentState      int           //当前的状态机状态
	recvHeartBeatTime int64         //上次收到心跳的时间

	// Your data here (2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	currentTerm int        //服务器最后一次知道的任期号
	votedFor    int        //当前获得选票的候选人ID
	log         []LogEntry //日志条目,包含指令和任期号

	committedIndex int //已知最大的已经被提交的日志条目索引值
	lastApplied    int //已经被状态机消费的日志条目索引

	//Leader的状态
	nextIndex  []int //需要发送给他的下一个日志条目索引值
	matchIndex []int //已经复制给他的日志最高索引值
}

// return currentTerm and whether this server
// believes it is the leader.
func (this *Raft) GetState() (int, bool) {
	var term int
	var isleader bool

	term = this.currentTerm
	isleader = this.currentState == RaftStateLeader

	return term, isleader
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (this *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(this.xxx)
	// e.Encode(this.yyy)
	// data := w.Bytes()
	// this.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (this *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   this.xxx = xxx
	//   this.yyy = yyy
	// }
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	Term         int //候选人任期号
	CandidateId  int //请求选票的后选人ID
	LastLogIndex int //候选人最后日志条目的索引值
	LastLogTerm  int //候选人最后日志条目的任期号
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	Term        int  //当前任期号, 以便于候选人更新自己的任期号
	VoteGranted bool //候选人赢得此选票时为真
}

//
// example RequestVote RPC handler.
//
func (this *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	fmt.Printf("%s Raft:%d RecvVotes:{%v}\n",
		getTimeOffset(), this.myId, *args)
	//接收者实现：
	//    如果term < currentTerm返回 false （5.2 节）
	if args.Term < this.currentTerm || (args.Term == this.currentTerm && this.currentState == RaftStateCandidate) {
		fmt.Printf("%s Raft:%d CandidateId:%d Term:%d, MyTerm:%d\n",
			getTimeOffset(), this.myId, args.CandidateId, args.Term, this.currentTerm)
		reply.Term = this.currentTerm
		reply.VoteGranted = false
		return
	}
	if args.Term > this.currentTerm {

		//this.votedFor = args.CandidateId
		//this.currentTerm = args.Term
		this.currentState = RaftStateFollower

		reply.Term = this.currentTerm
		reply.VoteGranted = true
		fmt.Printf("%s Raft:%d CandidateId:%d Term:%d, MyTerm:%d, change to follower\n",
			getTimeOffset(), this.myId, args.CandidateId, args.Term, this.currentTerm)
		return
	}
	//    如果 votedFor 为空或者为 candidateId，并且候选人的日志至少和自己一样新，那么就投票给他（5.2 节，5.4 节）
	if (this.votedFor == -1 || this.votedFor == args.CandidateId) && (args.LastLogIndex >= this.committedIndex) {
		go fmt.Printf("%s Raft:%d change to follower, MyTerm:%d, RemoteTerm:%d\n",
			getTimeOffset(), this.myId, this.currentTerm, args.Term)
		this.votedFor = args.CandidateId
		this.currentTerm = args.Term
		this.currentState = RaftStateFollower

		reply.VoteGranted = true
		return
	}
	fmt.Printf("ERROR\n")
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (this *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := this.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

type AppendEntriesRequest struct {
	Term         int        //领导人任期号
	LeaderId     int        //领导人的 Id，以便于跟随者重定向请求
	PrevLogIndex int        //新的日志条目紧随之前的索引值
	PrevLogTerm  int        //prevLogIndex 条目的任期号
	Entries      []LogEntry //准备存储的日志条目（表示心跳时为空；一次性发送多个是为了提高效率）
	LeaderCommit int        //领导人已经提交的日志的索引值
}

type AppendEntriesReply struct {
	Term         int  //当前的任期号，用于领导人去更新自己
	Success      bool //跟随者包含了匹配上 prevLogIndex 和 prevLogTerm 的日志时为真
	CurrentIndex int  //follower日志的最大Index
}

func (this *Raft) AppendEntries(request *AppendEntriesRequest, reply *AppendEntriesReply) {
	//    如果 term < currentTerm 就返回 false （5.1 节）
	if request.Term < this.currentTerm {
		reply.Term = this.currentTerm
		reply.Success = false
		return
	}

	//心跳
	if (this.currentState == RaftStateCandidate && this.currentTerm <= request.Term) || (this.currentState == RaftStateLeader && this.currentTerm < request.Term) {
		fmt.Printf("%s Raft:%d change %s state to follower, MyTerm:%d, NewTerm:%d\n",
			getTimeOffset(), this.myId, getStateString(this.currentState), this.currentTerm, reply.Term)
		this.currentState = RaftStateFollower
		this.recvHeartBeatTime = getMilliSeconds()
		this.votedFor = request.LeaderId
		this.currentTerm = request.Term

		reply.Term = this.currentTerm
		reply.Success = true
		if len(request.Entries) <= 0 {
			return
		}
	}
	//心跳
	if request.Entries == nil || len(request.Entries) <= 0 {
		fmt.Printf("%s Raft:%d recvHeartBeat:%d, LeaderCommit:%d\n",
			getTimeOffset(), this.myId, getMilliSeconds(), request.LeaderCommit)
		this.recvHeartBeatTime = getMilliSeconds()
		reply.Term = this.currentTerm
		reply.Success = true

		this.tryFollowerApplyLog(request.LeaderCommit)
		return
	}
	this.currentTerm = request.Term
	//TODO
	//	//    如果 leaderCommit > commitIndex，令 commitIndex 等于 leaderCommit 和 新日志条目索引值中较小的一个

	//Term         int        //领导人任期号
	//LeaderId     int        //领导人的 Id，以便于跟随者重定向请求
	//PrevLogIndex int        //新的日志条目紧随之前的索引值
	//PrevLogTerm  int        //prevLogIndex 条目的任期号
	//Entries      []LogEntry //准备存储的日志条目（表示心跳时为空；一次性发送多个是为了提高效率）
	//LeaderCommit int        //领导人已经提交的日志的索引值
	fmt.Printf("%s Raft:%d, recvAppendLog:{LeaderID:%d, PrevIndex:%d, PrevTerm:%d, Entries[0]:%v, LeaderCommit:%d, LeaderTerm:%d}\n",
		getTimeOffset(), this.myId, request.LeaderId, request.PrevLogIndex, request.PrevLogTerm, request.Entries[0].Data, request.LeaderCommit, request.Term)

	index := len(this.log) - 1
	term := -1
	if index > 0 {
		term = this.log[index].Term
	}

	//    如果日志在 prevLogIndex 位置处的日志条目的任期号和 prevLogTerm 不匹配，则返回 false （5.3 节）
	if request.PrevLogIndex > index {
		fmt.Printf("%s Raft:%d recvAppendLog fail, Index:%d != %d, Term:%d != %d\n",
			getTimeOffset(), this.myId, index, request.PrevLogIndex, term, request.PrevLogTerm)
		this.recvHeartBeatTime = getMilliSeconds()
		reply.Term = this.currentTerm
		reply.Success = false
		reply.CurrentIndex = index
		return
	}

	if request.PrevLogIndex >= 0 {
		index = request.PrevLogIndex
		term = this.log[index].Term
	}
	//    如果已经存在的日志条目和新的产生冲突（索引值相同但是任期号不同），删除这一条和之后所有的 （5.3 节）
	if index != request.PrevLogIndex || term != request.PrevLogTerm {
		fmt.Printf("%s Raft:%d recvAppendLog fail, Index:%d != %d, Term:%d != %d\n",
			getTimeOffset(), this.myId, index, request.PrevLogIndex, term, request.PrevLogTerm)
		this.recvHeartBeatTime = getMilliSeconds()
		reply.Term = this.currentTerm
		reply.Success = false
		reply.CurrentIndex = index - 1
		return
	}

	//    附加日志中尚未存在的任何新条目
	if index != len(this.log)-1 {
		//删除本地的日志, 用Master的日志覆盖本地的
		if index == -1 {
			this.log = make([]LogEntry, 0)
		} else {
			this.log = this.log[0 : index+1]
			fmt.Printf("%s Raft:%d delete log, Index:%d\n",
				getTimeOffset(), this.myId, index)
		}
	}
	var lastIndex = len(this.log) - 1
	var lastTerm = -1
	for _, item := range request.Entries {
		this.log = append(this.log, item)
		lastIndex += 1
		lastTerm = item.Term
	}
	fmt.Printf("%s Raft:%d recvAppendLog success, Index:%d, Term:%d\n",
		getTimeOffset(), this.myId, lastIndex, lastTerm)
	this.recvHeartBeatTime = getMilliSeconds()
	reply.Term = this.currentTerm
	reply.Success = true

	this.tryFollowerApplyLog(request.LeaderCommit)
	return
}

func (this *Raft) tryFollowerApplyLog(masterIndex int) {
	var log = this.log
	var currentCommitIndex = len(log) - 1
	for index := this.lastApplied + 1; index <= masterIndex && index <= currentCommitIndex; index += 1 {
		this.lastApplied += 1
		var entry = log[index]
		this.applyChan <- ApplyMsg{
			CommandValid: true,
			Command:      entry.Data,
			CommandIndex: index,
		}
		fmt.Printf("%s Raft:%d follower ApplyLog, Index:%d, Data:{%v}, LastApplied:%d\n",
			getTimeOffset(), this.myId, index, entry.Data, this.lastApplied)
	}
}

func (this *Raft) sendAppendEntries(server int, request *AppendEntriesRequest, reply *AppendEntriesReply) bool {
	ok := this.peers[server].Call("Raft.AppendEntries", request, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (this *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true
	this.mutex.Lock()
	defer this.mutex.Unlock()

	term, isLeader = this.GetState()

	if isLeader {
		this.log = append(this.log, LogEntry{
			Term: this.currentTerm,
			Data: command,
		})
		//从0开始
		index = len(this.log) - 1
		this.committedIndex = index

		fmt.Printf("%s Raft:%d AppendLog, Index:%d, Data:{%v}\n", getTimeOffset(), this.myId, index, command)
		return index, term, isLeader
	}

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (this *Raft) Kill() {
	atomic.StoreInt32(&this.dead, 1)
	// Your code here, if desired.
}

func (this *Raft) killed() bool {
	z := atomic.LoadInt32(&this.dead)
	return z == 1
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[myId]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	raft := &Raft{}
	raft.peers = peers
	raft.persister = persister
	raft.myId = me
	raft.votedFor = -1
	raft.recvHeartBeatTime = 0
	raft.loop = false

	if len(peers) == 3 {
		raft.ServerIDS = serverIds3
	} else if len(peers) == 5 {
		raft.ServerIDS = serverIds5
	}
	var _ = raft.ServerIDS[0]
	fmt.Printf("%s MakeRaft:%d\n", getTimeOffset(), raft.myId)

	raft.applyChan = applyCh
	raft.log = make([]LogEntry, 0)
	raft.lastApplied = -1
	raft.committedIndex = -1

	var length = len(peers)
	//follower的日志
	raft.matchIndex = make([]int, length)
	raft.nextIndex = make([]int, length)
	for index := 0; index < length; index += 1 {
		//已经复制给follower最大的日志
		raft.matchIndex[index] = -1
		//下次要发给follower的日志
		raft.nextIndex[index] = -1
	}
	// Your initialization code here (2B, 2C).
	// initialize from state persisted before a crash
	raft.readPersist(persister.ReadRaftState())

	go raft.startLoop()
	return raft
}

const RaftStateFollower = 1
const RaftStateCandidate = 2
const RaftStateLeader = 3

const TimeOutMilliSeconds = 150
const SleepMilliSecondsMin = 0
const SleepMilliSecondsMax = 90

var ServerStartTime = getMilliSeconds()

func getSleepMilliSeconds() int {
	return (int)(rand.Int31n(SleepMilliSecondsMax-SleepMilliSecondsMin)) + SleepMilliSecondsMin + 1
}

func getTimeOffset() string {
	var t = float64(getMilliSeconds()-ServerStartTime) / 1000
	return fmt.Sprintf("%07.3f", t)
}

func getMilliSeconds() int64 {
	return time.Now().UnixNano() / int64(time.Millisecond)
}

func getStateString(state int) string {
	if state == RaftStateFollower {
		return "follower"
	}
	if state == RaftStateCandidate {
		return "candidate"
	}
	if state == RaftStateLeader {
		return "leader"
	}
	return "unknown"
}

//follower状态:
//450ms周期性的检测上次收到心跳的时间, 如果没有收到心跳, 则进入candidate状态
func (this *Raft) follower() {
	var currentTime = getMilliSeconds()
	if this.recvHeartBeatTime+TimeOutMilliSeconds < currentTime {
		this.currentState = RaftStateCandidate
		fmt.Printf("%s Raft:%d Follower TimeOut, change to Candidate, LastHeadtBeat:%d, CurrentMilliSeconds:%d, Diff:%d, Term:%d\n",
			getTimeOffset(), this.myId, this.recvHeartBeatTime, currentTime, currentTime-this.recvHeartBeatTime, this.currentTerm)
	} else {
		fmt.Printf("%s Raft:%d Follower, Term:%d\n",
			getTimeOffset(), this.myId, this.currentTerm)
	}
}

var serverIds3 = [][]int{{1, 2}, {0, 2}, {0, 1}}
var serverIds5 = [][]int{{1, 2, 3, 4}, {0, 2, 3, 4}, {0, 1, 3, 4}, {0, 1, 2, 4}, {0, 1, 2, 3}}

//candidate状态:
//睡150-300ms
//如果收到了投票请求, 那么响应, 接着进入follower状态
//投自己一票, 发起投票请求, 超时之前收到一半以上票数, 则进入leader状态
//否则接着睡眠重新投票
func (this *Raft) candidate() {
	fmt.Printf("%s Raft:%d Candidate\n", getTimeOffset(), this.myId)

	this.votedFor = -1
	var sleepTime = getSleepMilliSeconds()
	time.Sleep(time.Duration(sleepTime) * time.Millisecond)

	if this.currentState != RaftStateCandidate {
		fmt.Printf("%s Raft:%d candidate change to %s\n", getTimeOffset(), this.myId, getStateString(this.currentState))
		return
	}

	var beginTime = getMilliSeconds()
	var sendServers = this.ServerIDS[this.myId]
	var votedCount = 1
	var failVotedCount = 0

	var beginVoteTerm = this.currentTerm
	this.votedFor = this.myId

	var requestVoteArgs = &RequestVoteArgs{
		Term:         this.currentTerm + 1,
		CandidateId:  this.votedFor,
		LastLogIndex: this.committedIndex,
		LastLogTerm:  0,
	}

	var lambda = func(serverId int) {
		var reply = &RequestVoteReply{
			Term: -1,
		}
		var result = this.sendRequestVote(serverId, requestVoteArgs, reply)
		if result && reply.VoteGranted {
			var replyTime = getMilliSeconds()

			if replyTime-beginTime > int64(TimeOutMilliSeconds-sleepTime) {
				failVotedCount += 1
				fmt.Printf("%s Raft:%d, SendRequestVote to Raft:%d, timeout, but success\n", getTimeOffset(), this.myId, serverId)
			} else {
				votedCount += 1
				fmt.Printf("%s Raft:%d, SendRequestVote to Raft:%d, vote success, Term:%d\n",
					getTimeOffset(), this.myId, serverId, this.currentTerm)
			}

		} else if !reply.VoteGranted {
			fmt.Printf("%s Raft:%d, SendRequestVote to Raft:%d failed, MyTerm:%d, RemoteTerm:%d\n",
				getTimeOffset(), this.myId, serverId, this.currentTerm, reply.Term)
			if reply.Term >= 0 && reply.Term >= this.currentTerm {
				this.currentState = RaftStateFollower
				fmt.Printf("%s Raft:%d, change to follower\n", getTimeOffset(), this.myId)
			}
			if reply.Term == -1 && result {
				fmt.Printf("ERROR RPC Result:%v, Raft:%d, ServerID:%d\n", result, this.myId, serverId)
			}
			failVotedCount += 1
		} else {
			fmt.Printf("\n")
		}

		if votedCount >= len(this.ServerIDS)/2+1 && this.currentState == RaftStateCandidate {
			this.currentState = RaftStateLeader
			this.currentTerm = beginVoteTerm + 1
			//增加一个noop消息
			this.Start(nil)
			time.Sleep(time.Millisecond)
			//这边矫正所有的follower的日志Index
			//原始算法里面, 假设了follower和master之间的差别不是很大
			//所以从最高位的日志里面去复制, 不行的话回退一格, 最终会收敛
			currentIndex := len(this.log) - 1
			var length = len(this.nextIndex)
			for i := 0; i < length; i += 1 {
				this.nextIndex[i] = currentIndex
				this.matchIndex[i] = currentIndex
			}
			fmt.Printf("%s Raft:%d VotedCount:%d, Change to leader, Term:%d\n",
				getTimeOffset(), this.myId, votedCount, this.currentTerm)
			go this.leader()
		}
		if failVotedCount >= 2 {
			this.votedFor = -1
			fmt.Printf("%s Raft:%d, FailVotedCount:%d, ResetState\n", getTimeOffset(), this.myId, failVotedCount)
		}
	}
	for i := 0; i < len(sendServers); i += 1 {
		var serverId = sendServers[i]
		go lambda(serverId)
	}
}

//leader状态:
//450ms周期性的向其他人发送heartbeat, 超过一半以上成员及时响应则维持leader状态
//否则超时进入follower状态
func (this *Raft) leader() bool {
	if this.currentState != RaftStateLeader {
		fmt.Printf("%s Raft:%d not leader, exit loop\n", getTimeOffset(), this.myId)
		return false
	}
	fmt.Printf("%s Raft:%d leader, CommitIndex:%d, LastApplied:%d, Term:%d\n",
		getTimeOffset(), this.myId, len(this.log)-1, this.lastApplied, this.currentTerm)

	var sendServers = this.ServerIDS[this.myId]
	var lambda = func(serverId int) {
		//Term         int    //领导人任期号
		//LeaderId     int    //领导人的 Id，以便于跟随者重定向请求
		//PrevLogIndex int    //新的日志条目紧随之前的索引值
		//PrevLogTerm  int    //prevLogIndex 条目的任期号
		//Entries      []byte //准备存储的日志条目（表示心跳时为空；一次性发送多个是为了提高效率）
		//LeaderCommit int    //领导人已经提交的日志的索引值

		var currentCommit = this.committedIndex
		var nextIndex = this.nextIndex[serverId]
		var prevLogIndex = nextIndex - 1
		var prevLogTerm = -1
		if prevLogIndex < 0 {
			prevLogIndex = -1
		}

		if nextIndex > 0 {
			if nextIndex >= len(this.log) {
				nextIndex = len(this.log)
			}
			prevLogIndex = nextIndex - 1
			prevLogTerm = this.log[prevLogIndex].Term
		}
		var entries = make([]LogEntry, 0)
		for index := nextIndex; index <= currentCommit && index < len(this.log); index += 1 {
			if index < 0 {
				continue
			}
			entries = append(entries, this.log[index])
		}

		var request = &AppendEntriesRequest{
			Term:         this.currentTerm,
			LeaderId:     this.myId,
			PrevLogIndex: prevLogIndex,
			PrevLogTerm:  prevLogTerm,
			Entries:      entries,
			LeaderCommit: this.lastApplied,
		}

		var reply = &AppendEntriesReply{}
		this.sendAppendEntries(serverId, request, reply)
		if !reply.Success && reply.Term > this.currentTerm {
			this.currentState = RaftStateFollower
			//this.currentTerm = reply.Term
			fmt.Printf("%s Raft:%d leader change to follower, my Term:%d, recv Term:%d\n",
				getTimeOffset(), this.myId, this.currentTerm, reply.Term)
			return
		}
		if !reply.Success && reply.Term == this.currentTerm {
			//TODO, 这边需要处理follower日志和master对不上的情况
			fmt.Printf("%s Raft:%d, follower:%d's log Index:%d\n",
				getTimeOffset(), this.myId, serverId, reply.CurrentIndex)
			this.matchIndex[serverId] = reply.CurrentIndex
			this.nextIndex[serverId] = reply.CurrentIndex
		}
		if reply.Success && len(request.Entries) > 0 {
			this.nextIndex[serverId] = currentCommit + 1
			this.matchIndex[serverId] = currentCommit
			fmt.Printf("%s Raft:%d ServerID:%d, NextIndex:%d, MatchIndex:%d\n",
				getTimeOffset(), this.myId, serverId, this.nextIndex[serverId], this.matchIndex[serverId])

			//这边几计算一下最大的CommitLog, 然后Apply
			//下一次同步给follower之后, follower也开始Apply
			this.onFollowerCommitLog(serverId, currentCommit)
		}
	}
	var length = len(sendServers)
	for i := 0; i < length; i += 1 {
		var serverId = sendServers[i]
		go lambda(serverId)
	}
	return true
}

func (this *Raft) onFollowerCommitLog(serverId int, followerCurrentCommit int) {
	this.mutex.Lock()
	defer this.mutex.Unlock()
	//这边需要看所有的follower最大的日志Commit, 然后去看什么可以提交
	var log = this.log
	var masterCommit = len(this.log)
	var lastApply = this.lastApplied
	var servers = this.ServerIDS[this.myId]
	var maxCurrentCommit = followerCurrentCommit
	for {
		var count = 1 //master本来就算1个
		for i := 0; i < len(servers); i += 1 {
			var followerCommit = this.matchIndex[servers[i]]
			if followerCommit >= maxCurrentCommit {
				count += 1
			}
		}
		if count >= len(this.ServerIDS)/2+1 {
			break
		} else {
			maxCurrentCommit -= 1
		}
		if maxCurrentCommit <= masterCommit {
			fmt.Printf("%s Raft:%d, follower:%d index:%d, MatserCommitIndex:%d\n",
				getTimeOffset(), this.myId, serverId, followerCurrentCommit, masterCommit)
			return
		}
	}
	fmt.Printf("%s Raft:%d Master OnFollowerCommitLog, ServerID:%d CommitLogIndex:%d, MaxCurrentCommit:%d\n",
		getTimeOffset(), this.myId, serverId, followerCurrentCommit, maxCurrentCommit)
	for index := lastApply + 1; index <= maxCurrentCommit; index += 1 {
		this.lastApplied += 1
		var command = log[index]
		this.applyChan <- ApplyMsg{
			CommandValid: true,
			Command:      command.Data,
			CommandIndex: index,
		}
		fmt.Printf("%s Raft:%d Master ApplyLog, Index:%d, Data:{%v}, LastApplied:%d\n",
			getTimeOffset(), this.myId, index, command.Data, this.lastApplied)
	}
}

func (this *Raft) tryStartHeartBeatLoop() {
	this.loopMutex.Lock()
	defer this.loopMutex.Unlock()
	if this.loop {
		return
	}
	this.loop = true
	go func() {
		for this.loop {
			this.loop = this.leader()
			time.Sleep(time.Duration(TimeOutMilliSeconds/4) * time.Millisecond)
		}
		this.loopMutex.Lock()
		defer this.loopMutex.Unlock()
		this.loop = false
	}()
}

func (this *Raft) startLoop() {
	this.currentState = RaftStateFollower
	for {
		time.Sleep(time.Duration(TimeOutMilliSeconds) * time.Millisecond)

		switch this.currentState {
		case RaftStateFollower:
			this.follower()
			break
		case RaftStateCandidate:
			go this.candidate()
			break
		case RaftStateLeader:
			this.tryStartHeartBeatLoop()
			break
		}
	}
}
